package com.cloud.basic.TcpAndUdp;

import lombok.extern.slf4j.Slf4j;
import org.bouncycastle.ocsp.Req;

import java.io.IOException;
import java.net.*;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @program: basic
 * @description: 数据报文发送：
 *                 1.发送消息线程负责发送，发送后将消息放入容器中等待应答，
 *                 2.接受线程接收应答，并发送消息给接收端自己已收到信息，从容器中匹配后删除
 *                 3.重发线程负责重发，未收到应答的消息，发送3次后移出
 * @author: Cloud
 * @create: 2021/10/11 16:42:03
 */
@Slf4j
public class DatagramSend {

    // 本地要发送的地址对象
    private SocketAddress localAddress;

    // 发送的socket对象
    private DatagramSocket datagramSender;

    // 发送的目标地址
    private SocketAddress remoteAddress;

    // 本地缓存已发送的消息，map key 为消息id，value为消息本身
    private Map<Integer, RequestMessage> msgQueue = new ConcurrentHashMap<>();

    public static void main(String[] args) throws SocketException {
        new DatagramSend();
    }

    public DatagramSend() throws SocketException {
        localAddress = new InetSocketAddress("127.0.0.1", 13000);
        datagramSender = new DatagramSocket(localAddress);
        remoteAddress = new InetSocketAddress("127.0.0.1",14000);

        // 启动线程
        startSendThread();

        // 接收线程-接收应答
        startRecvResponseThread();

        // 重发线程-负责重发
        startReSendThread();

    }

    @SuppressWarnings("all")
    public void startReSendThread(){
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    resendMsg();
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
    }

    /**
     * 功能描述: 判断map中的消息，如果超过3秒未收到应答则重发。
     *
     * @return void
     * @author cloud
     * @date  2021/10/11 17:08
     */
    public void resendMsg(){
        Set<Integer> set = msgQueue.keySet();
        Iterator<Integer> iterator = set.iterator();
        while (iterator.hasNext()) {
            Integer key = iterator.next();
            RequestMessage requestMessage = msgQueue.get(key);

            // 如果重发三次，则移出
            if (requestMessage.getSendCount()>=3) {
                iterator.remove();
                System.out.println("发送端--检测到丢失的消息：" + requestMessage);
            }
            // 当前时间
            long startTime = System.currentTimeMillis();

            if (startTime-requestMessage.getLastSendTime()>3000 && requestMessage.getSendCount() < 3) {
                byte[] buffer = requestMessage.toByte();
                DatagramPacket datagramPacket = new DatagramPacket(buffer, buffer.length, requestMessage.getRemoteAddr());
                try {
                    datagramSender.send(datagramPacket);
                    requestMessage.setSendCount(requestMessage.getSendCount()+1);
                    System.out.println("客户端重新发送消息:"+requestMessage);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private void startRecvResponseThread(){
        new Thread(new Runnable() {
            @Override
            public void run() {
                RecvResponse();
            }
        }).start();
    }

    public void RecvResponse() {
        System.out.println("接收端-接收应答线程启动！");
        while (true) {
            byte[] recvData = new byte[100];
            DatagramPacket datagramPacket = new DatagramPacket(recvData,recvData.length);
            try {
                datagramSender.receive(datagramPacket);
                RequestMessage requestMessage = new RequestMessage(recvData);
                int reqId = requestMessage.getId();
                RequestMessage requestMessage1 = msgQueue.get(new Integer(reqId));
                if (requestMessage1 != null) {
                    System.out.println("发送端-原来发送的数据："+requestMessage1);
                    System.out.println("接受的数据：" + requestMessage);
                    System.out.println("发送端-已收到接收端返回的信息："+new String(requestMessage.getData()));
                    msgQueue.remove(reqId);
                    // 发送端需要告诉接收端，返回的数据已经接收到了
                    // 发送的数据
                    byte[] msgData = (reqId + " 数据已收到").getBytes();
                    // 创建要发送的数据
                    RequestMessage sendMessage = new RequestMessage(reqId, recvData);

                    // 要发送的数据，将要发送的数据转为字节数组
                    byte[] buffer = sendMessage.toByte();
                    // 创建数据包，指定内容，指定目标地址
                    DatagramPacket packet = new DatagramPacket(buffer, buffer.length, remoteAddress);
                    // 发送数据包
                    datagramSender.send(packet);

                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    // 启动发送请求消息线程
    public void startSendThread() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                send();
            }
        }).start();
    }


    /*
     * 功能描述: 模拟发送消息
     *
     * @return void
     * @author cloud
     * @date  2021/10/21 11:41
     */
    private void send() {
        System.out.println("发送端-发送数据线程启动");
        // 确认机制 id从0开始
        int id = 0;
        // 模拟发送10请求
        while (id <= 10) {
            id++;
            byte[] bytes = (id + " hello").getBytes();
            // 创建要发送的对象
            RequestMessage sendMessage = new RequestMessage(bytes);

            // 要发送的数据，将要发送的数据转换为字节数组
            byte[] sendData = sendMessage.toByte();

            // 创建数据包，指定内容，指定地址
            DatagramPacket datagramPacket = new DatagramPacket(sendData, sendData.length, remoteAddress);
            // 发送数据
            try {
                datagramSender.send(datagramPacket);
                // 缓存当前发送的请求
                sendMessage.setSendCount(1);
                sendMessage.setRemoteAddr(remoteAddress);
                msgQueue.put(id,sendMessage);
                System.out.println("客户端-数据已发送，缓存："+sendMessage);
                Thread.sleep(1000);
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
